package com.kingsoft.dc.khaos.module.spark.preprocess.transform

import com.kingsoft.dc.khaos.{DataSetChannel, KhaosConf, KhaosContext, SchemaChannel}
import com.kingsoft.dc.khaos.innertype.Schema
import com.kingsoft.dc.khaos.metadata.{Dependency, KhaosStructField}
import com.kingsoft.dc.khaos.module.SparkModule
import com.kingsoft.dc.khaos.module.spark.metadata.preprocess.{PreprocessInfo, StrategyInfo}
import com.kingsoft.dc.khaos.util.Logging
import org.apache.spark.sql.DataFrame
import org.json4s.DefaultFormats
import org.json4s.jackson.JsonMethods.{compact, parse, render}

/**
  * Created by chengguangqing on 2019/6/24.
  */
class Transform(_mid: String,
             kc: KhaosContext) extends SparkModule with Logging{

  override var dependencies: Seq[Dependency] = Seq[Dependency]()
  override var targets: Seq[Dependency] = Seq[Dependency]()

  private var strategy: StrategyInfo = _

  private var transformStrategy: TransformStrategy = _
  private var module_id: String = _
  private var config: String = _

  /**
    * 节点逻辑处理的准备信息
    *
    * @param khaosConf : 组件配置信息
    * @param kc        : 组件上下文环境
    */
  override def prepare(khaosConf: KhaosConf,
                       kc: KhaosContext): this.type = {

    //根据_mid获取关于当前组件的配置信息
    val moduleConf = getModuleConfiguration() match {
      case Some(conf) => conf
      case None => throw new IllegalAccessException("conf is null")
    }
    implicit val formats = DefaultFormats
    val preprocessInfo = parse(moduleConf, true).extract[PreprocessInfo]
    strategy = preprocessInfo.strategy
    module_id = preprocessInfo.id
    val strategy_clazz = strategy.clazz
    config = compact(render(strategy.config))
    try {
      transformStrategy = Class.forName(strategy_clazz, true, Thread.currentThread().getContextClassLoader)
        .newInstance().asInstanceOf[TransformStrategy]
    } catch {
      case e: Exception => println(e.getStackTrace)
    }
    this
  }

  /**
    * 节点执行逻辑
    *
    * @param dataCollector :  数据采集器
    */
  override def execute(dataCollector: DataSetChannel): this.type = {
    val res = dependencies.isEmpty match {
      case true => transformStrategy.exec(kc, module_id, config, null,targets)
      case false => transformStrategy.exec(kc, module_id, config, dependencies, targets)
    }

    //增加算子多输出的情况
    val resSeq = res.asInstanceOf[Seq[(String,DataFrame)]]
    if(resSeq.size > 1){
      println()
      resSeq.foreach{ res =>
        val depend = targets.filter(_.targetSeq.equals(res._1)).head
        dataCollector.emit[DataFrame](Dependency(_mid,depend.sourceSeq, depend.targetNode, res._1), res._2)
      }
    }
    else{
      targets.foreach{ target =>
        dataCollector.emit[DataFrame](target, resSeq.head._2.asInstanceOf[DataFrame])
      }
    }
    this
  }

  /**
    * 声明输出schema
    * @param declarer :  输出声明器
    */
  override def declareOutputSchema(declarer: SchemaChannel): this.type = {
    val res = dependencies.isEmpty match {
      case true => transformStrategy.schema(kc, module_id,config, null)
      case false => transformStrategy.schema(kc,module_id,config, dependencies)
    }
    declarer.declare(Dependency(_mid), res.asInstanceOf[Schema])
    this
  }

  /**
    * 获取配置信息
    *
    * @return :  返回配置信息
    */
  override def getModuleConfiguration(): Option[String] = {
    kc.conf.getOption(_mid)
  }


  /**
    * 添加trans的信赖
    *
    * @param dependency : 依赖信息，每个组件的唯一组件
    */
  override def addDependencies(dependency: Dependency): this.type = {
    dependencies = dependencies :+ dependency
    this
  }


  def addTarget(dependency: Dependency):this.type ={
    targets = targets :+ dependency
    this
  }




}
